package my.thesis.bolts.computationsReliable;

import java.util.HashMap;
import java.util.Map;

import my.thesis.utilities.JoinFieldsTupleSentiment;
import my.thesis.utilities.JoinFieldsTupleSentimentReliable;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class JoinPosNegSentimentReliable extends BaseRichBolt
{
	private static final long serialVersionUID = 1L;
	private HashMap<Long, JoinFieldsTupleSentimentReliable> tweets;
	private OutputCollector collector;
	
	public void prepare(Map stormConf, TopologyContext context,OutputCollector collector) 
	{
		tweets=new  HashMap<Long, JoinFieldsTupleSentimentReliable>();
		this.collector=collector;	
	}

	public void execute(Tuple tuple) 
	{
		Long id = tuple.getLongByField("id");
		String country=tuple.getStringByField("country");
		
		if (tuple.contains("pos_score"))
		{
		    Integer pos = tuple.getIntegerByField("pos_score");
		    
		    if (this.tweets.containsKey(id))
		    {
		    	JoinFieldsTupleSentimentReliable partial= this.tweets.get(id);
		    	
		        if(partial.getPos()!=null)
		        {
		        	     //duplicate tuple
		        		//do nothing
		        }
		        else if(partial.getNeg()!=null)
		        {
		        		partial.addToList(tuple);
		        		int score = partial.getNeg() + pos;
		        		collector.emit(partial.getAnchors(),new Values(id, score, country));
		        		for(Tuple t : partial.getAnchors())
		        		{
		        			collector.ack(t);
		        		}
		        		
		        		tweets.remove(id);
		        }
		    }	       				          	       
		    else
		    {
		    	JoinFieldsTupleSentimentReliable posonly = new JoinFieldsTupleSentimentReliable(pos,null);
		    	posonly.addToList(tuple);
		    	this.tweets.put(id, posonly);
		    }
		 }
		
		 if (tuple.contains("neg_score"))
		 {
		    Integer neg = tuple.getIntegerByField("neg_score");
		    
		    if (this.tweets.containsKey(id))
		    {
		    	JoinFieldsTupleSentimentReliable partial= this.tweets.get(id);
		    	
		        if(partial.getNeg()!=null)
		        {
		        	    //duplicate tuple
		        		//do nothing
		        }
		        else if(partial.getPos()!=null)
		        {
		        		partial.addToList(tuple);
		        		int score = partial.getPos() + neg;
		        		
		        		collector.emit(partial.getAnchors(),new Values(id, score, country));
		        		for(Tuple t : partial.getAnchors())
		        		{
		        			collector.ack(t);
		        		}
		        		
		        		tweets.remove(id);
		        }	       				          	       
		    }
		    else
		    {
		    	JoinFieldsTupleSentimentReliable negonly = new JoinFieldsTupleSentimentReliable(null,neg);
		    	negonly.addToList(tuple);
		    	this.tweets.put(id, negonly);
		    }
		}
	}
		
	public void declareOutputFields(OutputFieldsDeclarer declarer) 
	{
		declarer.declare(new Fields("id","score","country"));	
	}

}
